/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  RocketMqProducerAdaptor.java
 * @Package com.rtf.repair.mq.rocketmq.producer
 * @Description: rocketmq 生产者适配器
 * @author: 司福林
 * @date:   2020-10-15 19:57:58
 * @version V1.0
 * @Copyright: 2020 RTF. All rights reserved.
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ----------------------------------------------------------------------------------
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.rocketmq.producer;

import java.io.Serializable;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.BeanInstantiationException;
import org.springframework.beans.BeanUtils;
import org.springframework.util.Assert;

import com.rtf.framework.mq.common.CallBack;
import com.rtf.framework.mq.common.MQException;
import com.rtf.framework.mq.common.MessageProducer;
import com.rtf.framework.mq.utils.ByteUtils;

/**
 *
 * @Description: rocketmq 生产者适配器
 * @ClassName:  RocketMqProducerAdaptor
 * @author: sfl
 * @date:   2020-10-15 19:57:58
 * @since:  v1.0
 */
public class RocketMqProducerAdaptor implements MessageProducer {

	private MQProducer producer;

	private MessageQueueSelector selectMessageQueueByHash = new SelectMessageQueueByHash();

	private MessageQueueSelector selectMessageQueueByPartion = new MessageQueueSelector() {

		@Override
		public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

			return mqs.get((Integer) arg);
		}

	};

	public RocketMqProducerAdaptor(MQProducer producer, String queueSelector) {
		Assert.notNull(producer, "Rocketmq producer is null");
		this.producer = producer;
		if (!StringUtils.isBlank(queueSelector)) {
			try {
				selectMessageQueueByHash = (MessageQueueSelector) BeanUtils
						.instantiateClass(RocketMqProducerAdaptor.class.getClassLoader().loadClass(queueSelector));
			} catch (BeanInstantiationException | ClassNotFoundException e) {
				throw new MQException("", e);
			}
		}
	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg) {
		this.sendMsg(topic, null, key, msg);
	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg, CallBack callBack) {
		this.sendMsg(topic, null, key, msg, callBack);
	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg, Integer parttion) {
		this.sendMsg(topic, null, key, msg, parttion);

	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg, Integer parttion, CallBack callBack) {
		this.sendMsg(topic, null, key, msg, parttion, callBack);

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg) {
		this.doSend(this.buildMessage(topic, key, msg, tag), selectMessageQueueByHash, key);

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg, CallBack callBack) {
		this.doSend(this.buildMessage(topic, key, msg, tag), selectMessageQueueByHash, key, callBack);

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg, Integer parttion) {
		this.doSend(this.buildMessage(topic, key, msg, tag), selectMessageQueueByPartion, parttion);

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg, Integer parttion, CallBack callBack) {
		this.doSend(this.buildMessage(topic, key, msg, tag), selectMessageQueueByPartion, parttion, callBack);

	}

	private Message buildMessage(String topic, String key, Serializable msg, String tag) {

		return new Message(topic, tag, key, ByteUtils.toBytes(msg));
	}

	private void doSend(Message message, MessageQueueSelector messageQueueSelector, Object partionParam) {
		try {
			SendResult result = producer.send(message, messageQueueSelector, partionParam);
			if (SendStatus.SEND_OK != result.getSendStatus()) {
				throw new MQException("rocketmq send failed");
			}
		} catch (Exception e) {
			throw new MQException("rocketmq send failed", e);
		}
	}

	private void doSend(Message message, MessageQueueSelector messageQueueSelector, Object partionParam,
			CallBack callBack) {
		try {

			producer.send(message, messageQueueSelector, partionParam, new RocketMqProducerCallBack(callBack));

		} catch (Exception e) {
			throw new MQException("rocketmq send failed", e);
		}
	}

	@Override
	public Object getProducer() {
		// TODO Auto-generated method stub
		return this.producer;
	}
}
